/**
  data-stream.c - Created by Timothy Morey on 5/14/2012
*/


/*****************************************************************************
 * System includes:
 *****************************************************************************/

#include <netcdf.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#ifdef __WINDOWS__
#include <Windows.h>
#pragma warning(disable:4996)
#else
#include <pthread.h>
#endif


/*****************************************************************************
 * Local includes:
 *****************************************************************************/

#include "client.h"
#include "data-schema.h"
#include "data-stream.h"
#include "err.h"
#include "logger.h"
#include "netcdf-utils.h"


/*****************************************************************************
 * Local definitions:
 *****************************************************************************/

typedef struct
{
  Client* Client;
  char DataSetName[256];
  char VariableName[32];
  int StreamThreadAlive;
  int KeepStreamThreadAlive;
  int ManageSelf;

#ifdef __WINDOWS__
  HANDLE StreamThread;
#else
  pthread_t StreamThread;
#endif
} DataStream;

#define MAX_STREAMS 32

static DataStream* g_DataStreams[MAX_STREAMS];

int StopStream(DataStream* stream);
int StreamLoop(DataStream* stream);

#ifdef __WINDOWS__
DWORD WINAPI StreamThreadEntryPoint(LPVOID arg);
#else
void* StreamThreadEntryPoint(void* arg);
#endif


/*****************************************************************************
 * Implementation of server.h functions:
 *****************************************************************************/

int CreateDataStream(const char* destination, int port,
                     const char* dataset, const char* varname, int* datastream)
{
  int retval = NVN_NOERR;
  int i = 0;
  int pos = -1;
  DataStream* ds = 0;
  char cport[8];

  if(!destination || !dataset || !varname || !datastream)
  {
    retval = NVN_EINVARGS;
  }

  if(NVN_NOERR == retval)
  {
    for(i = 0; i < MAX_STREAMS; i++)
    {
      if(0 == g_DataStreams[i])
      {
        g_DataStreams[i] = (DataStream*)1; // make sure nobody takes our space
        pos = i;
				break;
      }
    }

    if(pos == -1)
    {
      retval = NVN_EMAXOBJS;
    }
  }

  if(NVN_NOERR == retval)
  {
    ds = (DataStream*)malloc(sizeof(DataStream));
    memset(ds, 0, sizeof(DataStream));
    ds->KeepStreamThreadAlive = 1;
    ds->ManageSelf = 0;
    strcpy(ds->DataSetName, dataset);
    strcpy(ds->VariableName, varname);
    sprintf(cport, "%d", port);
    ConnectClient(destination, cport, &ds->Client);

#ifdef __WINDOWS__

    ds->StreamThread = 
      CreateThread(0, 0, StreamThreadEntryPoint, ds, 0, 0);

    if(0 >= ds->StreamThread)
#else
    if(0 != pthread_create(&ds->StreamThread, 0, StreamThreadEntryPoint, ds))
#endif
    {
      WriteMessage(LogLevelError, "Failed to create thread for data stream");
      ds->KeepStreamThreadAlive = 0;
      StopStream(ds);
      retval = NVN_ETHREADFAIL;
    }
  }

  if(NVN_NOERR == retval)
  {
    g_DataStreams[pos] = ds;
    *datastream = pos;
  }

  return retval;
}

int DestroyDataStream(int stream)
{
  int retval = NVN_NOERR;

  if(stream >= 0 && stream < MAX_STREAMS && g_DataStreams[stream])
  {
    retval = StopStream(g_DataStreams[stream]);
    g_DataStreams[stream] = 0;
  }
  else
  {
    retval = NVN_EINVARGS;
  }

  return retval;
}

int StartDataStream(const char* destination, int port, const char* dataset, const char* varname)
{
  int retval = NVN_NOERR;
  int ds = -1;

  retval = CreateDataStream(destination, port, dataset, varname, &ds);
  if(NVN_NOERR == retval)
    g_DataStreams[ds]->ManageSelf = 1;

  return retval;
}


/*****************************************************************************
 * Implementation of local functions:
 *****************************************************************************/

int StopStream(DataStream* stream)
{
  int retval = NVN_NOERR;

  if(stream)
  {
    if(stream->StreamThreadAlive)
    {
      stream->KeepStreamThreadAlive = 0;

#ifdef __WINDOWS__
      WaitForSingleObject(stream->StreamThread, INFINITE);
      CloseHandle(stream->StreamThread);
      stream->StreamThread = 0;
#else
      pthread_join(stream->StreamThread, 0);
#endif
    }

    if(stream->Client)
    {
      DestroyClient(stream->Client);
      stream->Client = 0;
    }

    free(stream);
  }
  else
  {
    retval = NVN_EINVARGS;
  }

  return retval;
}

int StreamLoop(DataStream* stream)
{
  int retval = 1;

  int ncstatus = NC_NOERR;
  int ncid = 0;
  int varid = 0;
  nc_type vartype;
  int ndims = 0;
  int dimids[NC_MAX_VAR_DIMS];
  int i = 0;
  int xdimid = -1, ydimid = -1, tdimid = -1;
  size_t xdimlen = 0, ydimlen = 0, tdimlen = 0;
  int xdimpos = -1, ydimpos = -1, tdimpos = -1;
  char* buf = 0;
  size_t buflen = 0;
  char logmsg[64];

  if(retval &&
    NC_NOERR != (ncstatus = nc_open(stream->DataSetName, 0, &ncid)))
  {
    sprintf(logmsg, "failed to nc_open %s with code %d", stream->DataSetName, ncstatus);
    WriteMessage(LogLevelError, logmsg);
    retval = 0;
  }

  if(retval &&
    NC_NOERR != (ncstatus = nc_inq_varid(ncid, stream->VariableName, &varid)))
  {
    sprintf(logmsg, "nc_inq_varid failed with %d", ncstatus);
    WriteMessage(LogLevelError, logmsg);
    retval = 0;
  }

  if(retval &&
    NC_NOERR != (ncstatus = nc_inq_var(ncid, varid, 0, &vartype, &ndims, dimids, 0)))
  {
    sprintf(logmsg, "nc_inq_var failed with %d", ncstatus);
    WriteMessage(LogLevelError, logmsg);
    retval = 0;
  }

  if(retval &&
    NC_NOERR != (ncstatus = nc_inq_unlimdim(ncid, &tdimid)))
  {
    sprintf(logmsg, "nc_inq_unlimdim failed with %d", ncstatus);
    WriteMessage(LogLevelError, logmsg);
    retval = 0;
  }

  if(retval)
  {
    for(i = 0; i < ndims; i++)
    {
      if(dimids[i] == tdimid)
      {
        tdimpos = i;
      }
			else if(tdimid == -1)
			{
				tdimid = dimids[i];
				tdimpos = i;
			}
      else if(xdimid == -1)
      {
        xdimid = dimids[i];
        xdimpos = i;
      }
      else
      {
        ydimid = dimids[i];
        ydimpos = i;
      }
    }

    if(xdimid == -1 || ydimid == -1 || tdimid == -1)
    {
      WriteMessage(LogLevelError, "failed to identify variable dimensions");
      retval = 0;
    }
  }

  if(retval &&
    (NC_NOERR != (ncstatus = nc_inq_dimlen(ncid, xdimid, &xdimlen)) ||
     NC_NOERR != (ncstatus = nc_inq_dimlen(ncid, ydimid, &ydimlen)) ||
     NC_NOERR != (ncstatus = nc_inq_dimlen(ncid, tdimid, &tdimlen))))
  {
    sprintf(logmsg, "nc_inq_dimlen failed with %d", ncstatus);
    WriteMessage(LogLevelError, logmsg);
    retval = 0;
  }

  if(retval && xdimlen > 0 && ydimlen > 0 && tdimlen > 0)
  {
    GetTypeSize(vartype, &buflen);

    if(buflen <= 0)
    {
      sprintf(logmsg, "unrecognized variable type: %d", vartype);
      WriteMessage(LogLevelError, logmsg);
      retval = 0;
    }

    buflen *= xdimlen;
    buflen *= ydimlen;
    buf = (char*)malloc(buflen);
  }

  if(retval && buf && buflen > 0)
  {
    size_t startslab[NC_MAX_DIMS];
    size_t countslab[NC_MAX_DIMS];

    countslab[xdimpos] = xdimlen;
    countslab[ydimpos] = ydimlen;
    countslab[tdimpos] = 1;

    for(i = 0; retval && (size_t)i < tdimlen; i++)
    {
      startslab[xdimpos] = 0;
      startslab[ydimpos] = 0;
      startslab[tdimpos] = i;

      if(NC_NOERR == (ncstatus = nc_get_vara(ncid, varid, startslab, countslab, buf)))
      {
				sprintf(logmsg, "Sending frame %d for %s", i, stream->VariableName);
				WriteMessage(LogLevelInfo, logmsg);

        if(NVN_NOERR != Send(stream->Client, buf, buflen))
				{
					WriteMessage(LogLevelError, "Error sending to client -- stopping data stream");
					retval = 0;
				}
      }
      else
      {
        sprintf(logmsg, "nc_get_vara failed with %d", ncstatus);
        WriteMessage(LogLevelError, logmsg);
        retval = 0;
      }
    }
  }

  return retval;
}

#ifdef __WINDOWS__
DWORD WINAPI StreamThreadEntryPoint(LPVOID arg)
#else
void* StreamThreadEntryPoint(void* arg)
#endif
{
  DataStream* ds = (DataStream*)arg;
  if(ds)
  {
    ds->StreamThreadAlive = 1;
    StreamLoop(ds);
    ds->StreamThreadAlive = 0;

    //if(ds->ManageSelf)
    //  TODO: cleanup
  }

  return 0;
}
